package com.sunday.common.mq.rocket.brave.study.apache.e3_Transactional_Message;

import com.sunday.common.mq.rocket.brave.study.apache.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.stream.Stream;

/**
 * 事务性消息
 * https://github.com/apache/rocketmq-spring/wiki/Transactional-Message
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class TransactionalMessageApi {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    private static final String topic = "your_topic";

    @GetMapping("send")
    public void send() {
        Stream.iterate(1, i -> i < 10, i -> i + 1).forEach(i -> {
            User user = User.builder().id(i).userName("john" + i).userAge(33 + i).build();
            Message<User> message = MessageBuilder.withPayload(user).build();
            rocketMQTemplate.sendMessageInTransaction(topic, message, Map.of("ptcg", "1996"));
        });
    }

    @Slf4j
    @Service
    @RocketMQMessageListener(topic = "your_topic", consumerGroup = "your_consumer_group_name", consumeThreadNumber = 1)
    public static class MessageExtConsumer implements RocketMQListener<MessageExt> {
        @Override
        public void onMessage(MessageExt message) {
            log.info("{} - {}", new String(message.getBody()), message);
        }
    }

}